#ifndef ACE_TCP_CONNECTION_H
#define ACE_TCP_CONNECTION_H

#include "ace/Svc_Handler.h"
#include "../io_session.h"
#include "ace/SOCK_Stream.h"
#include "../tcp_session.h"
#include "../io_filter_chain.h"
#include "../tcp_acceptor_config.h"
#include "../tcp_connection_manager.h"
#include "../tcp_message_context.h"
#include "../io_handler.h"
#include "../tcp_session_config.h"

#include <boost/bind.hpp>
#include <boost/tuple/tuple.hpp>


namespace vpm
{
	namespace mina
	{
		namespace detail
		{
			//so follower the relation AceTcpConnection (1)<---->(n) TcpSession 
			//it's AceTcpConnection's responsibility to create or delete the TcpSession
            template <class Session= TcpSession,class SessionConfig=TcpSessionConfig >
			class AceTcpConnection : public ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH> 
			{
				typedef ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH> super;
				enum 
				{
					TCP_DATA_BUFFER_LENGTH = 8192
				};
			public:
				AceTcpConnection():m_ioHandler(NULL)
				{
					assert(0); //don't call this construct, it's used by template 
				}

				AceTcpConnection(IoHandler & handler,TcpAcceptorConfig & acceptConfig)
                    :m_ioHandler(&handler),m_tcpAcceptorConfig(&acceptConfig)
				{	
                    Session *pSession = (Session*)m_ioHandler->template create_session<Session,SessionConfig>
                        (acceptConfig.get_session_config());

					m_ioSession.reset(pSession);
					m_ioHandler->sesion_created(*m_ioSession);

					IoSessionAccessor::attach_writter(*m_ioSession,boost::bind(&AceTcpConnection::send_data,this,_1));
					IoSessionAccessor::attach_reader(*m_ioSession,boost::bind(&AceTcpConnection::recv_data,this,_1));
					IoSessionAccessor::attach_closer(*m_ioSession,boost::bind(&AceTcpConnection::close_connection,this));
				}

				virtual ~AceTcpConnection()
				{
					//printf("Deconstruct AceTcpConnection %p\n",this); 
				}

				virtual int open(void * p)
				{
					if (super::open(p) == -1 )
					{
						return -1; 
					}

					ACE_INET_Addr peerAddr; 
					if (this->peer().get_remote_addr(peerAddr) == 0)
					{
                        printf("Connection from %s:%d\n",peerAddr.get_host_addr(), peerAddr.get_port_number());
					}

					//IoFilterChain & evtChain = m_tcpAcceptorConfig->get_event_chain();
					TcpMessageContext evtMsg(NULL,0); 
					//IoFilterChain::FilterAction act =  evtChain.process(evtMsg);

                    m_ioSession->set_connect_state(true); 
					m_ioHandler->session_opened(*m_ioSession);


					SessionConfig * pTcpSessionConfig = (SessionConfig*)(m_ioSession->get_session_config());
					m_bTcpMode = pTcpSessionConfig->get_tcp_mode();
					return 0; 
				}

				// Called when input is available from the client.                                                 
				virtual int handle_input (ACE_HANDLE handle = ACE_INVALID_HANDLE)
				{					
					//if it's the synch mode , don't fetch the data out.
					if (m_bTcpMode == SessionConfig::TCP_ASYNCH)
					{
						int recvLen = this->peer().recv(m_dataBuf , TCP_DATA_BUFFER_LENGTH ); 
						if (recvLen <=0 ) 
						{
                            printf("Connection closed \n"); 
							return -1; 
						}

						//process the message in the input chain 
						IoFilterChain & inChain = m_tcpAcceptorConfig->get_input_chain();
						TcpMessageContext context(m_dataBuf,recvLen); 
						inChain.process(context);
						m_ioHandler->message_received(*m_ioSession,context);

						//ACE_DEBUG ((LM_DEBUG,ACE_TEXT ("Received data length %d\n"),recvLen));
					}
					return 0; 
				}

				// Called when output is possible. 
				virtual int handle_output (ACE_HANDLE handle = ACE_INVALID_HANDLE)
				{
					return 0; 
				}

				// Called when this handler is removed from the ACE_Reactor. 
				virtual int handle_close (ACE_HANDLE handle,ACE_Reactor_Mask close_mask)
				{
                    m_ioSession->set_connect_state(false); 
					m_ioHandler->session_closed(*m_ioSession);
					return 0; 
				}

				int send_data(MessageContext & msg)
				{
					TcpMessageContext & tcpMsg = (TcpMessageContext&) msg;
					return this->send_data(tcpMsg.buffer,tcpMsg.length);		
				}

				virtual int send_data(const char * pData, int dataLen)
				{		
					return this->peer().send(pData,dataLen);
				}

				int recv_data(MessageContext & msg)
				{
					TcpMessageContext & tcpMsg = (TcpMessageContext&) msg;
					return this->recv_data(tcpMsg.buffer,tcpMsg.length);
				}

				virtual int recv_data(char * pBuf, int bufLen)
				{
					return this->peer().recv(pBuf,bufLen);
				}

			protected:
				virtual int close_connection()
				{
					this->close();
					this->shutdown();	
					return 0; 
				}

			private:
				char m_dataBuf[TCP_DATA_BUFFER_LENGTH]; 

				IoHandler * m_ioHandler; 
				TcpAcceptorConfig * m_tcpAcceptorConfig; 	
				std::auto_ptr<IoSession> m_ioSession;
				SessionConfig* m_sessionConfig;
				typename SessionConfig::TcpMode m_bTcpMode;
			};
#define TcpConnectionImpl AceTcpConnection
		}
	}
}

#endif // 


